import org.apache.flink.api.common.functions.FlatMapFunction
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

/**
 * ${description}
 *
 * @author Rubin
 * @version v1 2020/11/11 23:21
 */
object SocketTextStreamTest extends App {


  // Flink 程序的第一步是创建一个 StreamExecutionEnvironment 。
  // 这是一个入口类，可以用来设置参数和创建数据源以及提交任务。
  val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

  val text = env.socketTextStream("localhost", 8000)


  val windowCounts = text.flatMap(new FlatMapFunction[String, (String, Integer)]() {
    override def flatMap(value: String, out: Collector[(String, Integer)]): Unit = {
      for (word <- value.split("\\s")) {
        out.collect((word, 1))
      }
    }
  }).keyBy(0).timeWindow(Time.seconds(1),Time.seconds(1)).sum(1)



  // 将结果打印到控制台，注意这里使用的是单线程打印，而非多线程
  windowCounts.print.setParallelism(1)

  env.execute("Socket Window WordCount")
}

